package com.nl.pulsardemo.pool.config.produce;


import com.nl.pulsardemo.pool.config.commons.MyRouteMode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.pulsar.client.api.*;

import java.util.UUID;

import static com.nl.pulsardemo.pool.config.commons.Constants.SERVICE_URL;
import static com.nl.pulsardemo.pool.config.commons.Constants.TOPIC_NAME;

/**
 * @author : ZhaoY
 * @description :
 * @date : 2021-01-31 16:08
 **/
@Slf4j
//@Service
public class PooledProducerFactory implements PooledObjectFactory<Producer> {
    private PulsarClient pulsarClient;

    public PooledProducerFactory() {
//        try {
//            pulsarClient = PulsarClient.builder()
//                    .serviceUrl(SERVICE_URL)
//                    .build();
//        } catch (PulsarClientException e) {
//            log.info("PulsarClient初始化失败");
//            e.printStackTrace();
//        }
    }

    public PooledProducerFactory(PulsarClient pulsarClient) {
        this.pulsarClient = pulsarClient;
    }

    @Override
    public PooledObject<Producer> makeObject() throws Exception {
        return new DefaultPooledObject<Producer>(pulsarClient.newProducer(Schema.STRING)
                .topic(TOPIC_NAME)
                .enableBatching(true)//是否开启批量处理消息，默认true,需要注意的是enableBatching只在异步发送sendAsync生效，同步发送send失效。因此建议生产环境若想使用批处理，则需使用异步发送，或者多线程同步发送
                .compressionType(CompressionType.LZ4)//消息压缩（四种压缩方式：LZ4，ZLIB，ZSTD，SNAPPY），consumer端不用做改动就能消费，开启后大约可以降低3/4带宽消耗和存储（官方测试）
                .messageRouter(new MyRouteMode())//自定义消息路由
                .producerName(UUID.randomUUID().toString())//定义生产者名
                .create());
    }

    @Override
    public void destroyObject(PooledObject<Producer> pooledObject) throws Exception {
        pooledObject.getObject().close();

    }

    @Override
    public boolean validateObject(PooledObject<Producer> pooledObject) {
        return false;
    }

    @Override
    public void activateObject(PooledObject<Producer> pooledObject) throws Exception {

    }

    @Override
    public void passivateObject(PooledObject<Producer> pooledObject) throws Exception {

    }
}
